package com.atguigu.tingshu.search.service.impl;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.google.common.collect.Lists;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.repository.AlbumInfoRespository;
import com.atguigu.tingshu.search.service.SearchService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    AlbumInfoRespository albumInfoRespository;

    @Autowired
     AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    UserInfoFeignClient userInfoFeignClient;

    @Autowired
    CategoryFeignClient categoryFeignClient;

    private static ExecutorService executorService;

    @Autowired
    ElasticsearchClient elasticsearchClient;

    static {
        executorService = Executors.newFixedThreadPool(4);   // 创建一个线程 且未来这个线程池中会有四个线程
    }

    @SneakyThrows
    @Override
    public void OnSaleAlbum(Long albumId) {
//        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
//        albumInfoIndex.setId(albumId);
//        //专辑基本信息
//        Result<AlbumInfo>albumInfo  = albumInfoFeignClient.getAlbumInforbc(albumId);
//        Assert.notNull(albumInfo.getData());
//        AlbumInfo AlbumInfoData = albumInfo.getData();
//        albumInfoIndex.setAlbumTitle(AlbumInfoData.getAlbumTitle());
//        albumInfoIndex.setAlbumIntro(AlbumInfoData.getAlbumIntro());
//        albumInfoIndex.setCoverUrl(AlbumInfoData.getCoverUrl());
//        albumInfoIndex.setIncludeTrackCount(AlbumInfoData.getIncludeTrackCount());
//        albumInfoIndex.setIsFinished(AlbumInfoData.getIsFinished()+"");
//        albumInfoIndex.setPayType(AlbumInfoData.getPayType());
//        albumInfoIndex.setCreateTime(new Date());
//
//
//        //远程查询用户信息
//        Result<UserInfo>UserInfoData = userInfoFeignClient.getUserInfoByIdrpc(AlbumInfoData.getUserId());
//        Assert.notNull(UserInfoData.getData());
//        UserInfo userInfo = UserInfoData.getData();
//        albumInfoIndex.setAnnouncerName(userInfo.getNickname());
//
//        //远程调用分类1，2，3
//        Result<BaseCategoryView>categoryById123rpc = categoryFeignClient.getCategoryById123rpc(AlbumInfoData.getId());
//        Assert.notNull(categoryById123rpc.getData());
//        BaseCategoryView data = categoryById123rpc.getData();
//        albumInfoIndex.setCategory1Id(data.getCategory1Id());
//        albumInfoIndex.setCategory2Id(data.getCategory2Id());
//        albumInfoIndex.setCategory3Id(data.getCategory3Id());
//
//
//        List<AlbumAttributeValue> albumAttributeValueVoList = AlbumInfoData.getAlbumAttributeValueVoList();
//        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
//            List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList.stream().map(albumAttributeValue -> {
//                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
//                attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
//                attributeValueIndex.setValueId(albumAttributeValue.getValueId());
//                return attributeValueIndex;
//            }).collect(Collectors.toList());
//            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
//        }
//
//
//          //远程查询专辑的统计信息
//        Result<AlbumStatVo> albumInfoByIdrpc = albumInfoFeignClient.getAlbumInfoStartByIdrpc(albumId);
//        Assert.notNull(albumInfoByIdrpc.getData());
//        AlbumStatVo albumInfoByIdrpcdata = albumInfoByIdrpc.getData();
//        Integer playStatNum = albumInfoByIdrpcdata.getPlayStatNum();
//        Integer buyStatNum = albumInfoByIdrpcdata.getBuyStatNum();
//        Integer commentStatNum = albumInfoByIdrpcdata.getCommentStatNum();
//        Integer subscribeStatNum = albumInfoByIdrpcdata.getSubscribeStatNum();
//
//        albumInfoIndex.setPlayStatNum(albumInfoByIdrpcdata.getPlayStatNum());
//        albumInfoIndex.setSubscribeStatNum(albumInfoByIdrpcdata.getSubscribeStatNum());
//        albumInfoIndex.setBuyStatNum(albumInfoByIdrpcdata.getBuyStatNum());
//        albumInfoIndex.setCommentStatNum(albumInfoByIdrpcdata.getCommentStatNum());
//        //热度值
//        Double hotScore = playStatNum * 0.3 + buyStatNum * 0.3 + commentStatNum * 0.2 + subscribeStatNum * 0.2;
//       albumInfoIndex.setHotScore(hotScore);
//        AlbumInfoIndex save = albumInfoRespository.save(albumInfoIndex);
//        log.info(albumId + "专辑索引保存成功",save == null?"保存失败":"保存成功");

        Long beginTime = System.currentTimeMillis();
        // 1. 创建一个文档对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        albumInfoIndex.setId(albumId);

        Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInforbc(albumId);
        Assert.notNull(albumInfoResult.getData()); // albumInfoResult.getData()可能是空
        AlbumInfo albumInfoResultData = albumInfoResult.getData();
        albumInfoIndex.setAlbumTitle(albumInfoResultData.getAlbumTitle());
        albumInfoIndex.setAlbumIntro(albumInfoResultData.getAlbumIntro());
        albumInfoIndex.setCoverUrl(albumInfoResultData.getCoverUrl());
        albumInfoIndex.setIncludeTrackCount(albumInfoResultData.getIncludeTrackCount());
        albumInfoIndex.setIsFinished(albumInfoResultData.getIsFinished() + "");
        albumInfoIndex.setPayType(albumInfoResultData.getPayType());
        albumInfoIndex.setCreateTime(new Date());
        List<AlbumAttributeValue> albumAttributeValueVoList = albumInfoResultData.getAlbumAttributeValueVoList();
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            List<AttributeValueIndex> attributeValueIndexs = albumAttributeValueVoList.stream().map(albumAttributeValue -> {
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
                attributeValueIndex.setValueId(albumAttributeValue.getValueId());
                return attributeValueIndex;
            }).collect(Collectors.toList());
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexs);
        }

        // 2.2 远程查询用户信息
        Result<UserInfo> userInfoResult = userInfoFeignClient.getUserInfoByIdrpc(albumInfoResultData.getUserId());
        Assert.notNull(userInfoResult.getData());
        albumInfoIndex.setAnnouncerName(userInfoResult.getData().getNickname()); // 用户表的别名


        // 2.3 远程查询专辑微服务 获取到分类id(1 2 3)
        Result<BaseCategoryView> categoryViewResult = categoryFeignClient.getCategoryById123rpc(albumId);
        BaseCategoryView categoryViewResultData = categoryViewResult.getData();
        Assert.notNull(categoryViewResultData);
        albumInfoIndex.setCategory1Id(categoryViewResultData.getCategory1Id());
        albumInfoIndex.setCategory2Id(categoryViewResultData.getCategory2Id());
        albumInfoIndex.setCategory3Id(categoryViewResultData.getCategory3Id());

        // 2.4 远程查询专辑微服务 获取到播放量、订阅量、购买量、评论量
        Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumInfoStartByIdrpc(albumId);
        AlbumStatVo albumStatVoResultData = albumStatVoResult.getData();
        Assert.notNull(albumStatVoResultData);
        Integer playStatNum = albumStatVoResultData.getPlayStatNum();
        Integer buyStatNum = albumStatVoResultData.getBuyStatNum();
        Integer commentStatNum = albumStatVoResultData.getCommentStatNum();
        Integer subscribeStatNum = albumStatVoResultData.getSubscribeStatNum();
        albumInfoIndex.setPlayStatNum(playStatNum);
        albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
        albumInfoIndex.setBuyStatNum(buyStatNum);
        albumInfoIndex.setCommentStatNum(commentStatNum);
        //热度值
        Double hotScore = playStatNum * 0.3 + buyStatNum * 0.3 + commentStatNum * 0.2 + subscribeStatNum * 0.2;
        albumInfoIndex.setHotScore(hotScore);

        // 3.保存专辑文档对象到es
        AlbumInfoIndex saveStatus = albumInfoRespository.save(albumInfoIndex);
        Long endTime = System.currentTimeMillis();
        log.info("保存专辑：{}，状态:{},耗时：{}", albumId, saveStatus == null ? "失败" : "成功", endTime - beginTime); // tomcat线程池分配的线程


    }

//        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
//        albumInfoIndex.setId(albumId);
//        FutureTask<Long> longFutureTask = new FutureTask<Long>(new Callable<Long>() {
//            @Override
//            public Long call() throws Exception {
//                //专辑基本信息
//                Result<AlbumInfo> albumInfo = albumInfoFeignClient.getAlbumInforbc(albumId);
//                Assert.notNull(albumInfo.getData());
//                AlbumInfo AlbumInfoData = albumInfo.getData();
//                albumInfoIndex.setAlbumTitle(AlbumInfoData.getAlbumTitle());
//                albumInfoIndex.setAlbumIntro(AlbumInfoData.getAlbumIntro());
//                albumInfoIndex.setCoverUrl(AlbumInfoData.getCoverUrl());
//                albumInfoIndex.setIncludeTrackCount(AlbumInfoData.getIncludeTrackCount());
//                albumInfoIndex.setIsFinished(AlbumInfoData.getIsFinished() + "");
//                albumInfoIndex.setPayType(AlbumInfoData.getPayType());
//                albumInfoIndex.setCreateTime(new Date());
//                List<AlbumAttributeValue> albumAttributeValueVoList = AlbumInfoData.getAlbumAttributeValueVoList();
//                if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
//                    List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList.stream().map(albumAttributeValue -> {
//                        AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
//                        attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
//                        attributeValueIndex.setValueId(albumAttributeValue.getValueId());
//                        return attributeValueIndex;
//                    }).collect(Collectors.toList());
//                    albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
//                }
//                return AlbumInfoData.getUserId();
//            }
//        });
//        Thread thread = new Thread(longFutureTask);
//        thread.start();
//
//
//        Thread thread1 = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                Long userId = null;
//                try {
//                    userId = longFutureTask.get();
//                    //远程查询用户信息
//                    Result<UserInfo> UserInfoData = userInfoFeignClient.getUserInfoByIdrpc(userId);
//                    Assert.notNull(UserInfoData.getData());
//                    UserInfo userInfo = UserInfoData.getData();
//                    albumInfoIndex.setAnnouncerName(userInfo.getNickname());
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                } catch (Exception e) {
//                    throw new RuntimeException(e);
//                }
//
//            }
//        });
//        thread1.start();
//
//        Thread thread2 = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                //远程调用分类1，2，3
//                Result<BaseCategoryView> categoryById123rpc = categoryFeignClient.getCategoryById123rpc(albumId);
//                Assert.notNull(categoryById123rpc.getData());
//                BaseCategoryView data = categoryById123rpc.getData();
//                albumInfoIndex.setCategory1Id(data.getCategory1Id());
//                albumInfoIndex.setCategory2Id(data.getCategory2Id());
//                albumInfoIndex.setCategory3Id(data.getCategory3Id());
//            }
//        });
//        thread2.start();
//
//
//        Thread thread4 = new Thread(new Runnable() {
//            @Override
//            public void run() {
//                //远程查询专辑的统计信息
//                Result<AlbumStatVo> albumInfoByIdrpc = albumInfoFeignClient.getAlbumInfoStartByIdrpc(albumId);
//                Assert.notNull(albumInfoByIdrpc.getData());
//                AlbumStatVo albumInfoByIdrpcdata = albumInfoByIdrpc.getData();
//                Integer playStatNum = albumInfoByIdrpcdata.getPlayStatNum();
//                Integer buyStatNum = albumInfoByIdrpcdata.getBuyStatNum();
//                Integer commentStatNum = albumInfoByIdrpcdata.getCommentStatNum();
//                Integer subscribeStatNum = albumInfoByIdrpcdata.getSubscribeStatNum();
//
//                albumInfoIndex.setPlayStatNum(albumInfoByIdrpcdata.getPlayStatNum());
//                albumInfoIndex.setSubscribeStatNum(albumInfoByIdrpcdata.getSubscribeStatNum());
//                albumInfoIndex.setBuyStatNum(albumInfoByIdrpcdata.getBuyStatNum());
//                albumInfoIndex.setCommentStatNum(albumInfoByIdrpcdata.getCommentStatNum());
//                //热度值
//                Double hotScore = playStatNum * 0.3 + buyStatNum * 0.3 + commentStatNum * 0.2 + subscribeStatNum * 0.2;
//                albumInfoIndex.setHotScore(hotScore);
//            }
//        });
//        thread4.start();
//
//        thread.join();
//        thread1.join();
//        thread2.join();
//        thread4.join();
//
//
//
//        AlbumInfoIndex save = albumInfoRespository.save(albumInfoIndex);
//        log.info(albumId + "专辑索引保存成功",save == null?"保存失败":"保存成功");

    @Override
    public void batchAlbumDownSale() {
        albumInfoRespository.deleteAll();
        log.info("批量删除{}，成功");
    }

    @Override
    public void downSaleAlbum(Long albumId) {
        albumInfoRespository.deleteById(albumId);

        log.info("删除专辑：{}，成功", albumId);
    }

    @SneakyThrows
    @Override
    public List<Map<String, Object>> channel(Long c1Id) {
        ArrayList<Map<String, Object>> result = new ArrayList<>();
       Result<List<BaseCategory3>> baseCategory3Result= albumInfoFeignClient.getCategory3EntityByC1Id(c1Id);
        List<BaseCategory3> baseCategory3ResultData = baseCategory3Result.getData();
        Assert.notNull(baseCategory3ResultData);
        // 2.将这七个三级分类对象转成一个Map---<c3Id,BaseCategory3>
        Map<Long, BaseCategory3> c3IdAndEntityMap = baseCategory3ResultData
                .stream().collect(Collectors.toMap(BaseCategory3::getId, v -> v));
        // 3.将七个三级分类对象的三级分类id的java中Long类型转换为ES中能用的FieldValue类型
        List<FieldValue> esUseC3IdList = baseCategory3ResultData.stream().map(baseCategory3 -> FieldValue.of(baseCategory3.getId())).collect(Collectors.toList());
        // 4.开始检索
        // 4.1 构建SearchRequests对象
        SearchRequest searchRequest = SearchRequest
                .of(b -> b
                        .index("albuminfo")
                        .query(qb -> qb
                                .terms(tqb -> tqb
                                        .field("category3Id")
                                        .terms(tqfb -> tqfb
                                                .value(esUseC3IdList))))

                        .aggregations("category3IdAgg", ab -> ab
                                .terms(tab -> tab
                                        .field("category3Id")
                                        .size(10))
                                .aggregations("category3IdSubAgg", sab -> sab
                                        .topHits(thab -> thab.sort(sob -> sob
                                                        .field(fsb -> fsb
                                                                .field("hotScore")
                                                                .order(SortOrder.Desc)))
                                                .size(6)))));
        System.out.println("dsl语句" + searchRequest.toString());

        // 4.2 发送语句得到es的查询结果
        SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

        // 4.3 解析es中返回的数据 并且自定义封装前端要的数据

        // 4.3.1 获取三级分类id的字聚合对象
        Aggregate category3IdAgg = response.aggregations().get("category3IdAgg");
        // 4.3.2 获取这个聚合的真正类型
        LongTermsAggregate lterms = category3IdAgg.lterms();
        // 4.3.3 根据真正的聚合类型获取桶
        Buckets<LongTermsBucket> buckets = lterms.buckets();
        // 4.3.3 根据桶对象得到桶对象中的集合
        List<LongTermsBucket> bucketList = buckets.array();
        // 4.3.4 遍历桶集合
        for (LongTermsBucket longTermsBucket : bucketList) {
            long c3Id = longTermsBucket.key();
            Aggregate category3IdSubAgg = longTermsBucket.aggregations().get("category3IdSubAgg");
            TopHitsAggregate topHitsAggregate = category3IdSubAgg.topHits();
            List<Hit<JsonData>> hits = topHitsAggregate.hits().hits();
            ArrayList<AlbumInfoIndex> albumInfoIndicesList = new ArrayList<>();
            for (Hit<JsonData> hit : hits) {
                // 真正的数据
                JsonData source = hit.source();
                // 将es返回的数据模型转换前端要的数据模型
                AlbumInfoIndex albumInfoIndex = JSONObject.parseObject(source.toString(), AlbumInfoIndex.class);
                albumInfoIndicesList.add(albumInfoIndex);
            }
            // 构建返回对象
            HashMap<String, Object> map = new HashMap<>();
            map.put("list", albumInfoIndicesList);
            map.put("baseCategory3",c3IdAndEntityMap.get(c3Id));
            // 将三级对象放到集合中
            result.add(map);
        }

        return result;
    }

    @Override
    public Map<String, Object> albumInfoGet(Long albumId) {
        Map<String,Object>map = new HashMap<>();
        Result<AlbumInfo> albumInforbc = albumInfoFeignClient.getAlbumInforbc(albumId);
        Assert.notNull(albumInforbc);
        AlbumInfo albumInfoData = albumInforbc.getData();
        Long userId = albumInfoData.getUserId();
        map.put("albumInfo",albumInfoData);

        // 2.4 专辑对应的直播信息---远程查询
        Result<UserInfo> userInfoResult = userInfoFeignClient.getUserInfoByIdrpc(userId);
        UserInfo userInfoData = userInfoResult.getData();
        Assert.notNull(userInfoData);
        map.put("announcer", userInfoData);  // 放的是用户对象

        // 2.2 专辑的统计信息---远程查询
        Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumInfoStartByIdrpc(albumId);
        AlbumStatVo albumStatVoResultData = albumStatVoResult.getData();
        Assert.notNull(albumStatVoResultData);
        map.put("albumStatVo", albumStatVoResultData);

        // 2.3 专辑的分类信息---远程查询
        Result<BaseCategoryView> category123IdResult = categoryFeignClient.getCategoryById123rpc(albumId);
        BaseCategoryView category123IdData = category123IdResult.getData();
        Assert.notNull(category123IdData);
        map.put("baseCategoryView", category123IdData);
        return map;
    }


}
